package demo.delay;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.actuate.health.Status;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class KafkaDelayConsumerManager implements BeanPostProcessor, SmartInitializingSingleton {

    private Logger log = LoggerFactory.getLogger(KafkaDelayConsumerManager.class);

    /**
     * 延迟消费实例集合
     */
    private List<KafkaDelayConsumerContainer> consumerContainers = new ArrayList<>();

    private KafkaDelayConsumerDamonThread damonThread;

    @Resource
    private KafkaDelayConfig kafkaDelayConfig;

    @Resource
    private KafkaDltMessageHandler kafkaDltMessageHandler;

    /**
     * 健康检查
     *
     * @return
     */
    public Status health() {
        if (CollectionUtils.isEmpty(consumerContainers)) {
            return Status.UP;
        }
        Thread.State state = damonThread.getState();
        if (state == Thread.State.TERMINATED) {
            return Status.DOWN;
        }
        if (damonThread.getConsumersStatus().equals(KafkaDelayConsumerDamonThread.STATUS_UP)) {
            return Status.UP;
        }
        return Status.OUT_OF_SERVICE;
    }

    /**
     * 当spring中所有的对象都创建完成后会被调用 - init consumer threads
     */
    @Override
    public void afterSingletonsInstantiated() {
        if (!CollectionUtils.isEmpty(consumerContainers)) {
            Map<String, KafkaDelayConsumerThread> consumerThreadsMap = new HashMap<>();

            for (KafkaDelayConsumerContainer consumerContainer : consumerContainers) {
                KafkaConsumer<String, String> consumer = this.createConsumer(consumerContainer);
                consumerContainer.setConsumer(consumer);
                consumerContainer.setKafkaDltMessageHandler(kafkaDltMessageHandler);

                String topic = consumerContainer.getKafkaDelayListener().topic();
                KafkaDelayConsumerThread consumerThread = new KafkaDelayConsumerThread(consumerContainer);
                consumerThread.start();

                consumerThreadsMap.put(topic, consumerThread);
                log.info("KafkaDelayConsumerManager: start consume kafka topic {}", topic);
            }

            if (!CollectionUtils.isEmpty(consumerContainers)) {
                KafkaDelayConsumerDamonThread damon = new KafkaDelayConsumerDamonThread(consumerThreadsMap);
                damon.setDaemon(true);
                damon.start();
                log.info("KafkaDelayConsumerManager: start kafka delay daemon");

                damonThread = damon;
            }
        }
    }

    /**
     * 初始化后处理, 这里将添加了KafkaDelayListener注解的方法添加到consumerContainers中
     *
     * @param bean
     * @param beanName
     * @return
     */
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) {

        Class<?> targetClass = AopUtils.getTargetClass(bean);

        Map<Method, KafkaDelayListener> delayListenerMap = MethodIntrospector.selectMethods(targetClass,
                (MethodIntrospector.MetadataLookup<KafkaDelayListener>) method -> AnnotationUtils.findAnnotation(method, KafkaDelayListener.class));

        if (!CollectionUtils.isEmpty(delayListenerMap)) {

            for (Map.Entry<Method, KafkaDelayListener> entry : delayListenerMap.entrySet()) {
                consumerContainers.add(new KafkaDelayConsumerContainer(bean, entry.getKey(), entry.getValue()));
            }
        }
        return bean;
    }

    /**
     * @param consumerContainer
     * @return
     */
    private KafkaConsumer<String, String> createConsumer(KafkaDelayConsumerContainer consumerContainer) {

        Map<String, Object> map = kafkaDelayConfig.getConsumerProperties(consumerContainer.getKafkaDelayListener());

        return new KafkaConsumer<>(map);
    }


}